package org.joy.hpc.analyzer;

import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.Progressable;
import org.joy.db.IndexDB;
import org.joy.db.IndexEntry;
import org.joy.db.JoyDBOutputFormat;

public class IndexDBOutputFormat extends JoyDBOutputFormat<Text, HitWritable> {

	@Override
	public RecordWriter<Text, HitWritable> getRecordWriter(FileSystem fs,
			JobConf conf, String name, Progressable pro) throws IOException {
		// TODO Auto-generated method stub
		try {
			final IndexDB db = (IndexDB) env.openDB("org.joy.db.IndexDB");
			return new RecordWriter<Text, HitWritable>() {

				@Override
				public void close(Reporter reporter) throws IOException {
					// TODO Auto-generated method stub
					env.close();
				}

				@Override
				public void write(Text key, HitWritable val) throws IOException {
					// TODO Auto-generated method stub
					Integer i[] = new Integer[val.getPos().length];
					for (int j = 0; j < i.length; j++) {
						i[j] = ((IntWritable) val.getPos()[j]).get();
					}
					db.putEntry(new IndexEntry(key.toString().hashCode(), val
							.getUrl().hashCode(), val.getScore(), i));
				}

			};
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} 
		return null;
	}

}
